package com.bigdata.spark.mallapp_realtime

import java.lang
import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import com.bigdata.spark.util.{MyKafkaUtil, MyRedisUtil, OffsetManagerUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

object mall_log_ExactlyOnce_DA {
  def main(args: Array[String]): Unit = {

    val mall_log_Exactly_once_DA: SparkConf = new SparkConf().setMaster("local[*]").setAppName("mall_log_Exactly_once_DA")
    val ssc = new StreamingContext(mall_log_Exactly_once_DA, Seconds(10))


    var topic:String = "gmall_start_0523"
    var groupId = "ExactlyOnceDA"  //本陈程序的组id
    //即一条topic会被所有的group消费，属于一对多模式
    //只要group.id是全新的，就会从最新的的offset开始消费 只要不更改group.id，每次重新消费kafka，都是从上次消费结束的地方继续开始
    //consumer group下订阅的topic下的每个分区只能分配给某个group下的一个consumer(当然该分区还可以被分配给其他group)
    val offsetMap: Map[TopicPartition, Long] = OffsetManagerUtil.getOffset(topic, groupId)//获取


    var recordDStream: InputDStream[ConsumerRecord[String, String]] = null  //空DS
    //创建DStream

    if(offsetMap!=null && offsetMap.size >0){
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,offsetMap,groupId)  //
    }else{
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,groupId)   //
    }

    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]  //就是new Array[offsetRange](0)

    val offsetDStream: DStream[ConsumerRecord[String, String]] = recordDStream.transform {
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  //取到结束偏移量。

        rdd
      }
    }
    //- map是RDD中的每一个元素进行操作。
    //- mapPartition是对RDD的每一个分区的迭代器进行操作，返回的是迭代器。

    val DA: DStream[(String,Long)] = offsetDStream.mapPartitions {  //MapPartition（function）的function是RDD的分区进行操作，有多少个分区就会执行多少次，独立在每个分区上运行，所以mapPartitions效率比map高的多。
      kfiter => {                  //RDD迭代：
        val jedis: Jedis = MyRedisUtil.getJedisClient()
        val userList = new ListBuffer[(String,Long)]() //按照顺序保存登录信息，作为mapPartitions的返回值

        for (kfds <- kfiter) {             //对每个元素遍历
          val str: String = kfds.value()
          val startjson: JSONObject = JSON.parseObject(str)           //这是


          val ts: lang.Long = startjson.getLong("ts")

          val datestr: String = new SimpleDateFormat("yyyy-MM-dd").format(new Date(ts)) //获取登录时间
          val uid: String = startjson.getJSONObject("common").getString("uid") //获取uid
          println(uid)
          var EO_DAkey = "EO_DA:" + datestr //按照登录时间作为key。uid作为value。
          val isfirst = jedis.sadd(EO_DAkey, uid)

          if (jedis.ttl(EO_DAkey) < 0) { //查看还有多长时间失效，没设置就是-1.
            jedis.expire(EO_DAkey, 3600 * 24) //没设置失效时间的话，就设置失效时间。
          }

          //println(isfirst)
          if (isfirst != 0) {
            userList.append((uid, ts))
            println("新用户", startjson)
          }
        }
        jedis.close()
        //println(userList)
        userList.toIterator //返回List类型要转换过。要可迭代。给后面保存。这样就能按照时间顺序获得每个用户的首次登录时间。
      }
    }

//    DA.map(
//      Js =>{
//        //println(Js)
//        if(Js!=null){             //保存部分：只有有首次登录的用户出现，才会进入保存，一次保存一批次的用户。完成了实时日活+每个用户登录时间和顺序需求。
//          println("save,保存新用户",Js)
//        }
//
//      }
//    ).print()

    DA.foreachRDD(
      rdd=>{

        rdd.foreach(
          userts=>{
            if(userts == null){             //保存部分：只有有首次登录的用户出现，才会进入保存，一次保存一批次的用户。完成了实时日活+每个用户登录时间和顺序需求。
              //打开连接
              println("save,保存新用户",userts)
            }
          }

        )
        println("第一层")
        println(rdd,rdd.getClass)
        //提交偏移量到Redis中，以分区为单位。
        //println("提交偏移量到Redis中，以分区为单位。")
        OffsetManagerUtil.saveOffset(topic,groupId,offsetRanges)


      }

    )

    ssc.start()
    ssc.awaitTermination()




  }
}
